package com.rong.raft;

import java.io.Serializable;
import java.lang.reflect.Method;
import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import com.rong.raft.rpc.IRpcService;
import com.rong.raft.rpc.net.RpcClient;
import com.rong.raft.rpc.net.RpcServer;
import com.rong.raft.rpc.protocol.Constants;
import com.rong.raft.rpc.protocol.HeartbeatRequest;
import com.rong.raft.rpc.protocol.VoteRequest;
import com.rong.raft.rpc.protocol.VoteResponse;

/**
 * 
 * @desc leader选举器
 *
 * @author liurong
 *
 * @date 2019年10月21日
 */
public class LeaderSelector implements Serializable {

	private static final long serialVersionUID = 1L;

	private Random random = new Random();

	// 选举倒计时计时器
	private ScheduledExecutorService scheduledSeletorExecutorService;
	private ScheduledExecutorService scheduledHeartBeatService;
	private ScheduledFuture<?> scheduledSeletorFuture;
	private ScheduledFuture<?> scheduledHeartBeatFuture;

	// 当前节点node信息
	private NodeState nodeState;

	// 其他node的连接信息
	private static Map<String, RpcClient> rpcClientMap = new ConcurrentHashMap<String, RpcClient>();

	public LeaderSelector(NodeState nodeState) {
		this.nodeState = nodeState;

	}

	/**
	 * 封装RPC调用，目的为了RMI的重连机制
	 * 
	 * @param currentNodeState
	 * @param nodeId
	 * @param methodStr
	 * @param parameterType
	 * @param object
	 * @param returnType
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static <T> T sendToRemote(NodeState currentNodeState, String nodeId, String methodStr,
			Class<?> parameterType, Object object, Class<T> returnType) {
		RpcClient rpcClient = rpcClientMap.get(nodeId);
		try {
			IRpcService rpcService = rpcClient.getRpcService();
			Method method = rpcService.getClass().getMethod(methodStr, parameterType);
			return (T) method.invoke(rpcService, object);
		} catch (Exception e) {
			// RMI重连
			if (rpcClient.getIsStart()) {
				System.out.println("RMI重连！");
				String address = currentNodeState.getNodeMap().get(nodeId);
				rpcClient = new RpcClient(address);
				rpcClient.start();
				rpcClientMap.put(nodeId, rpcClient);
			}
			throw new RuntimeException(e);
		}
	}

	public void start() {
		synchronized (nodeState) {
			if (nodeState.getCurrNodeAddress() == null || nodeState.getCurrNodeAddress().trim().equals("")) {
				System.out.println("currNodeAddress不能为空！");
			}

			if (nodeState.getNodeMap() == null || nodeState.getNodeMap().size() == 0) {
				System.out.println("nodeMap不能为空！");
			}

			// 启动本地服务端
			RpcServer rpcServer = new RpcServer(nodeState.getCurrNodeAddress(), this);
			rpcServer.start();

			// 启动客户端服务
			Map<String, String> nodeMap = nodeState.getNodeMap();
			String address;
			RpcClient rpcClient;
			for (String nodeId : nodeMap.keySet()) {
				if (!nodeId.equals(nodeState.getSlefId())) {
					address = nodeMap.get(nodeId);
					rpcClient = new RpcClient(address);
					rpcClient.start();
					rpcClientMap.put(nodeId, rpcClient);
				}
			}

			// 启动选举定时器（时间150ms-300ms之间的随机值）
			scheduledSeletorExecutorService = Executors.newScheduledThreadPool(1);
			scheduledHeartBeatService = Executors.newScheduledThreadPool(1);

			this.resetSelectorTimer();
		}
	}

	private void selector() {
		synchronized (nodeState) {

			// leader节点不处理选举逻辑
			if (nodeState.getRole() == NodeRoleEnum.LEADER) {
				return;
			}

			// 判断当前node是否为follower节点
			if (nodeState.getRole() == NodeRoleEnum.FOLLOWER) {
				changeNodeToCandidate(nodeState);
			}

			// 发送投票
			sendVote(nodeState.getCurrTerm(), nodeState.getSlefId());

			if (nodeState.getRole() == NodeRoleEnum.CANDIDATE) {// 投票失败重新投票
				resetSelectorTimer();
			} else if (nodeState.getRole() == NodeRoleEnum.LEADER) {// 成为leader开启定时心跳
				resetHeartBeatTimer();
			}
		}
	}

	private void sendVote(long currTerm, String candidateId) {
		Map<String, String> nodeMap = nodeState.getNodeMap();
		Set<String> nodeKey = nodeMap.keySet();
		VoteRequest voteRequest = new VoteRequest();
		voteRequest.setCandidateId(candidateId);
		voteRequest.setCurrTerm(nodeState.nextTerm());

		List<String> agreeNodes = new ArrayList<String>();
		VoteResponse voteResponse;
		for (String nodeId : nodeKey) {
			try {
				System.out.println("开始发送投票：" + nodeId);
				voteRequest.setRemoteId(nodeId);
				if (nodeId.equals(candidateId)) {
					voteResponse = handleVote(voteRequest);
				} else {
					voteResponse = sendVoteToRemote(nodeId, voteRequest);
				}
				if (voteResponse == null) {
					continue;
				}

				if (Constants.VOTE_SUCCESS.equals(voteResponse.getCode())) {
					agreeNodes.add(voteResponse.getNodeId());
				}

				if (nodeState.isSelector(agreeNodes.size())) {
					nodeState.setLeaderId(nodeState.getSlefId());
					nodeState.setRole(NodeRoleEnum.LEADER);
					System.out.println(agreeNodes);
					System.out.println("成功选举成leader， nodeId=" + nodeState);
					break;
				}

			} catch (Exception e) {
				System.out.println("发送投票异常：" + e);
			}
		}
	}

	public VoteResponse handleVote(VoteRequest voteRequest) {
		try {
			VoteResponse voteResponse = new VoteResponse();
			System.out.println(voteRequest);
			System.out.println(nodeState);
			if (voteRequest.getCurrTerm() > nodeState.getCurrTerm()) {
				nodeState.setCurrTerm(voteRequest.getCurrTerm());
				nodeState.setCurrVoteFor(voteRequest.getCandidateId());

				voteResponse.setCode(Constants.VOTE_SUCCESS);
				voteResponse.setCurrVoteFor(voteRequest.getCandidateId());
				voteResponse.setNodeId(nodeState.getSlefId());

				cancelSelectorTimer();

			} else if (voteRequest.getCurrTerm() == nodeState.getCurrTerm()
					&& voteRequest.getCandidateId().equals(nodeState.getSlefId())) {
				nodeState.setCurrTerm(voteRequest.getCurrTerm());
				nodeState.setCurrVoteFor(voteRequest.getCandidateId());

				voteResponse.setCode(Constants.VOTE_SUCCESS);
				voteResponse.setCurrVoteFor(voteRequest.getCandidateId());
				voteResponse.setNodeId(nodeState.getSlefId());

				cancelSelectorTimer();

			}
			System.out.println(voteResponse);
			return voteResponse;
		} catch (Exception e) {
			System.out.println("处理投票结果报错！网路异常！");
		}
		return null;
	}

	private void sendHeartBeat() {
		synchronized (nodeState) {
			HeartbeatRequest heartbeatRequest = new HeartbeatRequest();
			heartbeatRequest.setLeaderId(nodeState.getLeaderId());
			heartbeatRequest.setCurrTerm(nodeState.getCurrTerm());
			for (String nodeId : rpcClientMap.keySet()) {
				new Thread(new Runnable() {
					@Override
					public void run() {
						try {
							sendToRemote(nodeState, nodeId, "sendHeartBeat", HeartbeatRequest.class, heartbeatRequest,
									Void.class);
//							this.getRpcClient(nodeId).getRpcService().sendHeartBeat(heartbeatRequest);
						} catch (Exception e) {
							System.out.println("发送心跳异常：" + e);
						}
					}
				}).start();
			}
		}
	}

	private void cancelHeartBeatTimer() {
		if (scheduledHeartBeatFuture != null && !scheduledHeartBeatFuture.isDone()) {
			scheduledHeartBeatFuture.cancel(true);
		}
	}

	private void resetHeartBeatTimer() {
		cancelHeartBeatTimer();

		scheduledHeartBeatFuture = scheduledHeartBeatService.scheduleAtFixedRate(new Runnable() {
			public void run() {
				sendHeartBeat();
			}

		}, 1000, 1000, TimeUnit.MILLISECONDS);
	}

	private void cancelSelectorTimer() {
		if (scheduledSeletorFuture != null && !scheduledSeletorFuture.isDone()) {
			scheduledSeletorFuture.cancel(true);
		}
	}

	private void resetSelectorTimer() {
		cancelSelectorTimer();

		scheduledSeletorFuture = scheduledSeletorExecutorService.schedule(new Runnable() {
			public void run() {
				selector();
			}

		}, 3000 + random.nextInt(1000), TimeUnit.MILLISECONDS);
	}

	/**
	 * 发送投票给远端
	 * 
	 * @param nodeId
	 * @param voteRequest
	 * @throws RemoteException
	 */
	private VoteResponse sendVoteToRemote(String nodeId, VoteRequest voteRequest) throws RemoteException {
		try {
			return sendToRemote(nodeState, nodeId, "sendVote", VoteRequest.class, voteRequest, VoteResponse.class);
//			return getRpcClient(nodeId).getRpcService().sendVote(voteRequest);
		} catch (Exception e) {
			System.out.println("发送投票给远端服务器 " + nodeId + " 报错！");
		}
		return null;
	}

	/**
	 * 将节点升级为candidate角色
	 * 
	 * @param nodeState
	 */
	private void changeNodeToCandidate(NodeState nodeState) {
		nodeState.setRole(NodeRoleEnum.CANDIDATE);
	}

	public void handleHeartBeart(HeartbeatRequest heartbeatRequest) {
		System.out.println("收到心跳" + heartbeatRequest);
		resetSelectorTimer();
		if (nodeState.getLeaderId() == null || nodeState.getLeaderId().equals("")) {
			nodeState.setLeaderId(heartbeatRequest.getLeaderId());
			nodeState.setCurrTerm(heartbeatRequest.getCurrTerm());
		} else if (!nodeState.getLeaderId().equals(heartbeatRequest.getLeaderId())) {// 如果leaderId不相同，说明出现了脑裂。取curTerm大的为新的leader
			if (heartbeatRequest.getCurrTerm() > nodeState.getCurrTerm()) {
				if (nodeState.getLeaderId().equals(nodeState.getSlefId())) {
					cancelHeartBeatTimer();
				}

				nodeState.setLeaderId(heartbeatRequest.getLeaderId());
				nodeState.setCurrTerm(heartbeatRequest.getCurrTerm());
			}
		}
	}

}
